Enforce co-partitioning for sort merge and symmetric hash joins#23480
Conversation
github-actions
Bot
added
core
| .map(|exec| Arc::new(exec) as Arc<dyn ExecutionPlan>) | ||
| } | ||
|
|
||
| fn hash_partitioned_exec( |
There was a problem hiding this comment.
using hash to prove that co-partitioning dows work fro an accepted partitioning type
| InputDistributionRequirements::co_partitioned(vec![ | ||
| Distribution::KeyPartitioned(left_expr), | ||
| Distribution::KeyPartitioned(right_expr), | ||
| ] | ||
| ]) |
There was a problem hiding this comment.
Add range-partitioning SLT coverage showing both joins repartitioned until their dedicated Range opt-in work lands.
Is it worth TODOs next to both of the uses of InputDistributionRequirements::co_partitioned linking to the relevant issues which will do it? Alternatively, would it be reasonably to inline that enabling? Otherwise it feels like this PR is nearly a noop.
There was a problem hiding this comment.
That's also my feeling after reviewing this PR, it seems to be doing close to nothing.
Rather than changing 2 LOC in symteric_hash_join.rs and sort_merge_join/mod.rs, and adding some tests demonstrating it still does not work, would it be too hard to actually make it work in this same PR?
There was a problem hiding this comment.
Ya that was prbably too small of changes per PR, I addressed this adding the support.
| 03)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true] | ||
| 04)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 | ||
| 05)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false | ||
| 06)----SortExec: expr=[range_key@0 ASC], preserve_partitioning=[true] | ||
| 07)------RepartitionExec: partitioning=Hash([range_key@0], 4), input_partitions=4 | ||
| 08)--------DataSourceExec: file_groups=<slt:ignore>, projection=[range_key, value], output_partitioning=Range([range_key@0 ASC], [(10), (20), (30)], 4), file_type=csv, has_header=false |
There was a problem hiding this comment.
Can you remind me how is it that, even if SortMergeJoinExec now requires both children to be co-partitioned, and the join is in the range_key, we are still seeing RepartitionExecs injected here? I bet this is tracked in an issue. Can we link the issue in the test description?
There was a problem hiding this comment.
it is because I did not opt into the range satisfaction since it isn't general yet with: allow_range_satisfaction_for_key_partitioning()
I didn't do this just because I wanted to do the other operators first but with others picking those up I guess there is no reason fro me not to 👍
| InputDistributionRequirements::co_partitioned(vec![ | ||
| Distribution::KeyPartitioned(left_expr), | ||
| Distribution::KeyPartitioned(right_expr), | ||
| ] | ||
| ]) |
There was a problem hiding this comment.
That's also my feeling after reviewing this PR, it seems to be doing close to nothing.
Rather than changing 2 LOC in symteric_hash_join.rs and sort_merge_join/mod.rs, and adding some tests demonstrating it still does not work, would it be too hard to actually make it work in this same PR?
gene-bordegaray
force-pushed
the
gene.bordegaray/2026/07/co-partitioned-join-requirements
branch
2 times, most recently
from
9eb6eb1 to
c17f438
Compare
gene-bordegaray
force-pushed
the
gene.bordegaray/2026/07/co-partitioned-join-requirements
branch
from
c17f438 to
80edd7c
Compare
| /// | ||
| /// `output_partitioning` must describe this plan's current output and have | ||
| /// the same number of partitions as the stream. | ||
| pub fn with_output_partitioning( |
There was a problem hiding this comment.
surprised this wasn't avaialble before, wondering if there was a reason but could not think of why 🤔
3 participants
Which issue does this PR close?
Distribution::KeyPartitionedrequirements before general Range satisfaction #23451Partitioning::Rangeinputs for sort merge joins #23478Partitioning::Rangeinputs for symmetric hash joins #23479Rationale for this change
Partition-index-aware joins require compatible input layouts. Compatible Range layouts can satisfy that without repartitioning.
What changes are included in this PR?
Are these changes tested?
Yes
Are there any user-facing changes?
Streaming table providers can declare output partitioning with
StreamingTable::with_output_partitioning().